// Fill out your copyright notice in the Description page of Project Settings.


#include "CurlHDFS.hpp"
#include "cybertron/core/UtilLz4.hpp"

#pragma warning(disable:4668) // curl: %s is not defined as a preprocessor macro, replacing with '0' for '#if/#elif'
#ifndef WIN32_LEAN_AND_MIN
  #define WIN32_LEAN_AND_MIN  // curl: indirectly include Windows.h
#endif

#define    CURL_STATICLIB 1
#include <curl/curl.h>

#include <fstream>
#include <map>

#include <string.h>

namespace static_curl {


static size_t streamReadCallback(char *buffer, size_t size, size_t nitems, std::istream* infile)
{
	infile->read(buffer, size * nitems);
	return infile->gcount();
}

static size_t CurlStreamCallback(char* ptr, size_t size, size_t nmemb, std::ostream* strm)
{
	strm->write(ptr, size * nmemb);
	return size * nmemb;
}

static int CurlDebugCallback(CURL* curl, curl_infotype type, char* data, size_t size, std::ostream* strm)
{
	switch (type) {
	case CURLINFO_HEADER_OUT:
	case CURLINFO_DATA_OUT:
	case CURLINFO_SSL_DATA_OUT:
	case CURLINFO_HEADER_IN:
	case CURLINFO_DATA_IN:
	case CURLINFO_SSL_DATA_IN:
		strm->write(data, size);
		break;
	}
	return 0;
}

FCurlResponse::FCurlResponse(void* curl)
	: _curl(curl)
	, _code(0)
{}

FCurlResponse::~FCurlResponse()
{}

int FCurlResponse::Code() const
{
	// Return the HTTP status code.
	return _code;
}

std::string FCurlResponse::Loc() const {
	return _location;
}

const std::string& FCurlResponse::ContentType() const
{
	return _contentType;
}

//json FCurlResponse::Json() const
//{
//	return json::parse(_stringOut.str());
//}

std::string FCurlResponse::PlainText() const
{
	return _stringOut.str();
}

bool FCurlResponse::IsCurlError() const
{
	// Error happens before the server returns any HTTP status code.
	return !_errstr.empty();
}

const std::string& FCurlResponse::CurlError() const
{
	// Return the error string from Curl.
	return _errstr;
}

std::string FCurlResponse::ErrorMessage() const
{
	if (IsCurlError())
	{
		// A client side error before reaching the server.
		return CurlError();
	}
	else
	{
		std::stringstream ss;
		ss << "Code: " << Code() << ". Response: " << PlainText();
		return ss.str();
	}
}

FCurlRequest::FCurlRequest()
	: _curl(curl_easy_init())
	, _headers(nullptr)
	, _method(Method::GET)
	, _response(_curl)
{
	curl_easy_setopt(_curl, CURLOPT_FOLLOWLOCATION, 1L);
	curl_easy_setopt(_curl, CURLOPT_NOSIGNAL, 1);
	curl_easy_setopt(_curl, CURLOPT_ACCEPT_ENCODING, "deflate");
	curl_easy_setopt(_curl, CURLOPT_TIMEOUT, 60L);
	curl_easy_setopt(_curl, CURLOPT_SSL_VERIFYPEER, 0L); // certs in UE4 ?
	curl_easy_setopt(_curl, CURLOPT_WRITEFUNCTION, CurlStreamCallback);
	curl_easy_setopt(_curl, CURLOPT_WRITEDATA, &(std::ostream&)_response._stringOut);
}

FCurlRequest::~FCurlRequest()
{
	if (_curl)
		curl_easy_cleanup(_curl);
	_curl = nullptr;

	if (_headers)
		curl_slist_free_all(_headers);
	_headers = nullptr;
}

FCurlRequest& FCurlRequest::AcceptEnc(const char* coding)
{
	curl_easy_setopt(_curl, CURLOPT_ACCEPT_ENCODING, coding);
	return *this;
}

FCurlRequest& FCurlRequest::FollowLoc(long follow_loc)
{
	curl_easy_setopt(_curl, CURLOPT_FOLLOWLOCATION, follow_loc);
	return *this;
}

FCurlRequest& FCurlRequest::Put(const std::string& endpoint, std::ifstream* infile)
{
	curl_easy_setopt(_curl, CURLOPT_UPLOAD, 1L);

	if (infile == nullptr) {
		curl_easy_setopt(_curl, CURLOPT_INFILESIZE, 0);
	//	_headers = curl_slist_append(_headers, "Transfer-Encoding:");
	}
	else {
		curl_easy_setopt(_curl, CURLOPT_INFILESIZE, -1);
		_headers = curl_slist_append(_headers, "Transfer-Encoding: chunked");
		curl_easy_setopt(_curl, CURLOPT_READFUNCTION, streamReadCallback);
		curl_easy_setopt(_curl, CURLOPT_READDATA, infile);
	}

	_endpoint = endpoint;
	_method = Method::PUT;
	return *this;
}

FCurlRequest& FCurlRequest::Get(const std::string& endpoint)
{
	curl_easy_setopt(_curl, CURLOPT_HTTPGET, 1L);
	_endpoint = endpoint;
	_method = Method::GET;
	return *this;
}

FCurlRequest& FCurlRequest::Get(const std::string& schemehostport, const std::string& path)
{
	return Get(schemehostport + path);
}

FCurlRequest& FCurlRequest::Post(const std::string& endpoint)
{
	curl_easy_setopt(_curl, CURLOPT_HTTPPOST, 1L);
	_endpoint = endpoint;
	_method = Method::POST;
	return *this;
}

FCurlRequest& FCurlRequest::Post(const std::string& schemehostport, const std::string& path)
{
	return Post(schemehostport + path);
}

FCurlRequest& FCurlRequest::Then(std::shared_ptr<FCurlExecutorBase>&& body)
{
	_then = std::move(body);
	return *this;
}

FCurlRequest& FCurlRequest::Then()
{
	_then = nullptr;
	return *this;
}


FCurlRequest& FCurlRequest::Catch(std::shared_ptr<FCurlExecutorBase>&& body)
{
	_catch = std::move(body);
	return *this;
}

FCurlRequest& FCurlRequest::Query(const std::string& name, const std::string& value)
{
	std::shared_ptr<char> nameStr(
		curl_easy_escape(_curl, name.c_str(), 0),
		[](char* p) {
			curl_free(p);
		});
	std::shared_ptr<char> valueStr(
		curl_easy_escape(_curl, value.c_str(), 0),
		[](char* p) {
			curl_free(p);
		});

	_querystring += _querystring.empty() ? "?" : "&";
	_querystring += nameStr.get();
	_querystring += "=";
	_querystring += valueStr.get();
	return *this;
}

FCurlRequest& FCurlRequest::Query()
{
	_querystring = "";
	return *this;
}

FCurlRequest& FCurlRequest::Stream(std::ostream& strm)
{
	// Redirect the output to a user-defined stream.
	curl_easy_setopt(_curl, CURLOPT_WRITEDATA, &(std::ostream&)strm);
	return *this;
}

//FCurlRequest& FCurlRequest::Json(const json& jsonObj)
//{
//	// Post the Json data.
//	if (_method == Method::POST)
//	{
//		// Json string as the post body.
//		curl_easy_setopt(_curl, CURLOPT_COPYPOSTFIELDS, jsonObj.dump().c_str());
//
//		// Set the appropriate header.
//		_headers = curl_slist_append(_headers, "Content-Type: application/json");
//	}
//	return *this;
//}

FCurlRequest& FCurlRequest::Debug(std::ostream& strm)
{
	curl_easy_setopt(_curl, CURLOPT_VERBOSE, 1L);
	curl_easy_setopt(_curl, CURLOPT_DEBUGFUNCTION, CurlDebugCallback);
	curl_easy_setopt(_curl, CURLOPT_DEBUGDATA, &(std::ostream&)strm);
	return *this;
}

FCurlResponse& FCurlRequest::Perform()
{
	// Get the final url for the HTTP request.
	const std::string url = _endpoint + _querystring;

	// Set any remaining curl options.
	curl_easy_setopt(_curl, CURLOPT_URL, url.c_str());

	// Headers
	if (_headers)
	{
		curl_easy_setopt(_curl, CURLOPT_HTTPHEADER, _headers);
	}

	// Initiate the HTTP request. Blocking.
	CURLcode curlcode = curl_easy_perform(_curl);

	if (curlcode == CURLE_OK)
	{
		// Success. Get the HTTP status code.
		long code = 0;
		curl_easy_getinfo(_curl, CURLINFO_RESPONSE_CODE, &code);

		// Build the response object.
		_response._code = static_cast<int>(code);
		_response._errstr.clear();

		if (code >= 200 && code < 400)
		{
			// Get the MIME of the response.
			const char* mime = nullptr;
			curl_easy_getinfo(_curl, CURLINFO_CONTENT_TYPE, &mime);
			_response._contentType = mime ? mime : "";

			if ((code / 100) == 3) {
				char *location = nullptr;
				curl_easy_getinfo(_curl, CURLINFO_REDIRECT_URL, &location);
				_response._location = std::string(location);
			}
			// Call the then handler.
			if (_then)
			{
				_then->Invoke(_response);
			}
		}
		else
		{
			// Call the catch handler.
			if (_catch)
			{
				_catch->Invoke(_response);
			}
		}
	}
	else
	{
		// Error happened before completing the HTTP request.
		_response._code = 400; // Bad Request ?
		_response._errstr.assign(curl_easy_strerror(curlcode));

		// Call the catch handler.
		if (_catch)
		{
			_catch->Invoke(_response);
		}
	}

	return _response;
}

} // namespace static_curl

  // parse_hdfs_url extract a http url which is a namenode webHDFS address , several hostnames and those corresponding ip addresss
  // for requesting datanode.

  // hdfs url example: 
  // hdfs[xhb-slave-1:10.2.35.118,xhb-slave-2:10.2.35.119,xhb-slave-3:10.2.35.4]://10.2.35.117:50070/webhdfs/v1/record/filename
  // [hostname:hostip] is a optional field, it may disapper if i become more familar with webHDFS
static bool parse_hdfs_url(const std::string& origin_url,
	std::string& http_url, std::map<std::string, std::string>& hosts) {
	auto pos_beg = origin_url.find('[');
	auto pos_end = origin_url.find(']');

	if (pos_beg != std::string::npos && pos_end != std::string::npos) {
		http_url = "http" + origin_url.substr(pos_end + 1);

		std::stringstream ss(origin_url.substr(pos_beg + 1, pos_end - pos_beg - 1));
		while (ss.good()) {
			std::string tmp;

			std::getline(ss, tmp, ',');

			auto tmp_pos = tmp.find(':');
			if (tmp_pos == std::string::npos)
				continue;
			hosts[tmp.substr(0, tmp_pos)] = tmp.substr(tmp_pos + 1);
		}
	}
	else if (pos_beg == std::string::npos && pos_end == std::string::npos) {
		http_url = "http" + origin_url.substr(strlen("hdfs"));
	}
	else {
		return false;
	}
	return true;
}

// transfrom_dn_url replace the part of data node url "http://xhb-slave-3:50075" to "http://10.2.35.117:50065"
static void transfrom_dn_url(std::string& dn_url,
	const std::map<std::string, std::string>& hosts) {
	for (auto it = hosts.begin(); it != hosts.end(); it++) {
		auto host_name = "http://" + it->first + ":";
		auto host_ip = "http://" + it->second + ":";

		auto pos = dn_url.find(host_name);
		if (pos != std::string::npos) {
			dn_url = host_ip + dn_url.substr(host_name.length());
			return;
		}
	}
}

// replace_blank replace the black space in url to %20
static void replace_blank(std::string &strBig, std::string strsrc, std::string strdst) {
	std::string::size_type pos = 0;
	std::string::size_type srclen = strsrc.size();
	std::string::size_type dstlen = strdst.size();
	while ((pos = strBig.find(strsrc, pos)) != std::string::npos) {
		strBig.replace(pos, srclen, strdst);
		pos += dstlen;
	}
}

bool UtilHDFSFile::downloadHDFSFileTo(
	const std::string& url,
	const std::string& dstFile) {
	std::string urlNoSpace = url;
	replace_blank(urlNoSpace, " ", "%20"); //handle space in file name case

	std::string nn_url;
	std::map<std::string, std::string> hosts;
	if (parse_hdfs_url(urlNoSpace, nn_url, hosts) == false)
		return false;

	std::string dn_url;
	static_curl::FCurlRequest hdfscli;

	hdfscli.FollowLoc(0).
		AcceptEnc(NULL).
		Get(nn_url)
		.Then([&](static_curl::FCurlResponse& res) {
		dn_url = res.Loc();
	})
		.Query(std::string("op"), std::string("OPEN"))
		.Perform();

	if (dn_url != "") {
		transfrom_dn_url(dn_url, hosts);
	}
	else
		return false;


	std::ofstream odfile;
	odfile.open(dstFile.c_str(), std::ofstream::out | std::ofstream::trunc | std::ofstream::binary);
	if (odfile) {
		hdfscli.Query()
			.Stream(odfile)
			.Then()
			.Get(dn_url).Perform();

		odfile.close();
		return true;
	}
	return false;
}

bool UtilHDFSFile::uploadHDFSFileTo(
	const std::string& srcFile,
	const std::string& url) {
	typedef std::shared_ptr<void> close_guard;

	std::ifstream infile;
	infile.open(srcFile.c_str(), std::ifstream::in | std::ifstream::binary);
	if (!infile)
		return false;
	close_guard ____(nullptr, [&](...) {infile.close(); });

	std::string urlNoSpace = url;
	replace_blank(urlNoSpace, " ", "%20"); //handle space in file name case

	std::string nn_url;
	std::map<std::string, std::string> hosts;
	if (parse_hdfs_url(urlNoSpace, nn_url, hosts) == false)
		return false;

	std::string dn_url;
	static_curl::FCurlRequest hdfscli;

	hdfscli.FollowLoc(0).
		AcceptEnc(NULL).
		Put(nn_url, nullptr).
		Then([&](static_curl::FCurlResponse& res) {
		dn_url = res.Loc();
	}).
		Query(std::string("op"), std::string("CREATE")).
		Perform();

	if (dn_url != "") {
		transfrom_dn_url(dn_url, hosts);
	}

	bool failed = false;
	hdfscli.Query().Then().Put(dn_url, &infile).Catch([&](static_curl::FCurlResponse& res) {
		failed = true;
		return false;
	}).Perform();

	return failed == false;
}

